package com.atguigu.edu.realtime.utils;

import edu.umd.cs.findbugs.annotations.Nullable;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.IOException;
import java.util.Properties;

/**
 * @ClassName MyKafkaUtil
 * @Description TODO
 * @Author$ 邢家俊
 * @Date 2023-5-4 22:59
 * @Version 1.0
 **/
public class MyKafkaUtil {
    public static final String STRING_BOOTSTRAP="hadoop103:9092,hadoop104:9092,hadoop105:9092";

    public static FlinkKafkaConsumer<String> getKafkaConsumer(String topic, String groupId){
        Properties propos = new Properties();
        propos.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,MyKafkaUtil.STRING_BOOTSTRAP);
        propos.setProperty(ConsumerConfig.GROUP_ID_CONFIG,groupId);
        propos.setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed");
        return new FlinkKafkaConsumer<String>(topic, new KafkaDeserializationSchema<String>() {
            @Override
            public boolean isEndOfStream(String nextElement) {
                return false;
            }
            @Override
            public String deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
                if( record.value() !=null){
                    return new String(record.value());
                }
                return null;
            }
            @Override
            public TypeInformation<String> getProducedType() {
                return TypeInformation.of(String.class);
            }
        }, propos);
    }

    public static String getTopicDbDDL(String groupid) {
        return "CREATE TABLE topic_db (\n" +
                "  `database` string,\n" +
                "  `table` string,\n" +
                "  `type` string,\n" +
                "  `data` Map<String,String>, \n" +
                "  `old` Map<string,string>,\n" +
                "  `ts` string,\n" +
                "  `proc_time` as proctime()\n" +
                ")"+getKafkaDDL("topic_db",groupid);
    }
    public static String getKafkaDDL(String topic ,String groupId){
        return "WITH ( 'connector' = 'kafka',\n" +
                "  'topic' = '"+topic+"',\n" +
                "  'properties.bootstrap.servers' = '"+STRING_BOOTSTRAP+"',\n" +
                "  'properties.group.id' = '"+groupId+"',\n" +
                "  'scan.startup.mode' = 'group-offsets',\n" +
                "  'format' = 'json'\n" +
                ")";
    }

    public static String getUpsertKafkaDDL(String topic){
       return  "WITH ( " +
                "  'connector' = 'upsert-kafka',\n" +
                "  'topic' = '"+topic+"',\n" +
                "  'properties.bootstrap.servers' = '"+STRING_BOOTSTRAP+"',\n" +
                "  'key.format' = 'json',\n" +
                "  'value.format' = 'json'\n" +
                ")";
    }
    public static FlinkKafkaProducer<String> getKafkaProducer(String topic) {

        //默认情况创建的FlinkKafkaProducer对象并不能保证生产一直性，因为semantic的值默认是AT_LEAST_ONCE
        //如果要想保证一致性，需要在创建对象的时候，指定semantic的值为EXACTLY_ONCE
        // FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<String>("hadoop102:9092,hadoop103:9092,hadoop104:9092","dirty_data",new SimpleStringSchema());

        Properties props = new Properties();
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, STRING_BOOTSTRAP);
        props.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 15 * 60 * 1000 + "");
        FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>("default_topic", new KafkaSerializationSchema<String>() {
            @Override
            public ProducerRecord<byte[], byte[]> serialize(String jsonStr, @Nullable Long timestamp) {
                return new ProducerRecord<byte[], byte[]>(topic, jsonStr.getBytes());
            }
        }, props, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
        return kafkaProducer;
    }
    public static <T>FlinkKafkaProducer<T> getKafkaProducerBySchema(KafkaSerializationSchema<T> kafkaSerializationSchema){
        Properties prope = new Properties();
        prope.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,STRING_BOOTSTRAP);
        prope.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 15 * 60 * 1000 + "");
        FlinkKafkaProducer<T> defaultTopic = new FlinkKafkaProducer<>("default_topic", kafkaSerializationSchema, prope, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
        return defaultTopic;
    }



}
